/*
 * Copyright (c) 2020 - present, Inspur Genersoft Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.inspur.edp.commonmodel.core.session.sessioncache;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.inspur.edp.cef.api.session.ICefSessionItem;
import com.inspur.edp.commonmodel.core.session.distributed.dac.FuncSessionIncrement;
import com.inspur.edp.commonmodel.core.session.json.SessionIncrementDeserializer;
import com.inspur.edp.commonmodel.core.session.json.SessionIncrementSerializer;
import com.inspur.edp.commonmodel.core.session.serviceinterface.SessionCacheInfoService;
import com.inspur.edp.commonmodel.core.session.tableentity.SessionCacheInfo;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.util.StringUtils;

import javax.sql.DataSource;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class SessionCacheInfoServiceImpl implements SessionCacheInfoService {
    private DataSource dataSource;
    private String tableName;
    private final NamedParameterJdbcTemplate template;
    private final DataSourceTransactionManager transactionManager;

    public SessionCacheInfoServiceImpl(DataSource dataSource, String tableName) {
        this.dataSource = dataSource;
        this.tableName = tableName;
        this.template = new NamedParameterJdbcTemplate(dataSource);
        this.transactionManager = new DataSourceTransactionManager(dataSource);
    }

    public List<SessionCacheInfo> getSessionCacheInfos(String sessionId, int version) {
        String sql = "select * from " + tableName + " where sessionId=:sessionId and version>:version order by version asc";
        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("sessionId", sessionId);
        paramMap.put("version", version);
        List<SessionCacheInfo> infos = template.query(sql, paramMap, new BeanPropertyRowMapper<>(SessionCacheInfo.class));
        for (SessionCacheInfo info : infos) {
            if (StringUtils.isEmpty(info.getSessionContent()))
                continue;
            info.setSessionChangeSet(DeSerializeChangeSet(info.getSessionContent()));
        }
        return infos;
    }

    @Override
    public void saveSessionCacheInfo(SessionCacheInfo info, Map<String, ICefSessionItem> items) {
        String sql = "insert into " + tableName + " (id, sessionId, version, sessionContent, createdOn) values (:id, :sessionId, :version, :sessionContent, :createdOn)";
        if (info.getSessionChangeSet() != null) {
            info.setSessionContent(SerializeChangeSet(info.getSessionChangeSet(), items));
        }
        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("id", info.getId());
        paramMap.put("sessionId", info.getSessionId());
        paramMap.put("version", info.getVersion());
        paramMap.put("sessionContent", info.getSessionContent());
        paramMap.put("createdOn", info.getCreatedOn());
        template.update(sql, paramMap);
    }

    @Override
    public void deleteSessionCacheInfos(String sessionId, int version) {
        String sql = "delete from " + tableName + " where sessionId=:sessionId and version<=:version";

        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("sessionId", sessionId);
        paramMap.put("version", version);
        template.update(sql, paramMap);
    }

    @Override
    public void clearSessionCacheInfos(String sessionId) {
        String sql = "delete from " + tableName + " where sessionId=:sessionId";
        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("sessionId", sessionId);
        template.update(sql, paramMap);
    }

    @Override
    public List<SessionCacheInfo> traversalSessionCacheInfos(Date date) {
        JdbcTemplate jdbcTemplate = template.getJdbcTemplate();
        try {
            //因为变更记录采用累加形式，同一个session可能有多条记录。
            //在此获取这些记录的目的是为了后续删除，因此同一个session只需查找到一条就可以，所以添加了version=0的条件
            String sql = "select * from " + tableName + " where createdOn > ? and version = 0 order by createdOn asc";
            jdbcTemplate.setMaxRows(500);
            List<SessionCacheInfo> infos = jdbcTemplate.query(sql, new BeanPropertyRowMapper<>(SessionCacheInfo.class), date);
            return infos;
        } finally {
            jdbcTemplate.setMaxRows(-1);
        }
    }

    @Override
    public void batchClear(List<String> sessionIds) {
        String sql = "delete from " + tableName + " where sessionid in (:sessionIds)";

        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("sessionIds", sessionIds);

        template.update(sql, paramMap);
    }

    private String SerializeChangeSet(FuncSessionIncrement increment, Map<String, ICefSessionItem> items) {
        ObjectMapper mapper = getMapper(items);
        try {
            String value = null;
            if (increment != null) {
                value = mapper.writeValueAsString(increment);
            }
            return value;
        } catch (JsonProcessingException | RuntimeException e) {
            throw new RuntimeException("ChangeSet序列化失败", e);
        }
    }

    private FuncSessionIncrement DeSerializeChangeSet(String increment) {
        ObjectMapper mapper = getMapper();
        try {
            FuncSessionIncrement vm = mapper.readValue(increment, FuncSessionIncrement.class);
            return vm;
        } catch (IOException e) {
            throw new RuntimeException("ChangeSet反序列化失败", e);
        }
    }

    private ObjectMapper getMapper() {
        ObjectMapper mapper = new ObjectMapper();
        SimpleModule module = new SimpleModule();
        module.addDeserializer(FuncSessionIncrement.class, new SessionIncrementDeserializer());
        mapper.registerModule(module);
        return mapper;
    }

    private ObjectMapper getMapper(Map<String, ICefSessionItem> items) {
        ObjectMapper mapper = new ObjectMapper();
        SimpleModule module = new SimpleModule();
        module.addSerializer(FuncSessionIncrement.class, new SessionIncrementSerializer(items));
        mapper.registerModule(module);
        return mapper;
    }

    @Override
    public boolean isLatestVersion(String sessionId, int version) {
        String sql = "select count(1) from " + tableName + " where sessionId=:sessionId and version>:version";

        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("sessionId", sessionId);
        paramMap.put("version", version);
        Integer num = template.queryForObject(sql, paramMap, Integer.class);
        return num != null && num <= 0;
    }
}
